package scribe.system.peer.scribe;

import common.peer.PeerAddress;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.network.Network;
import se.sics.kompics.timer.SchedulePeriodicTimeout;
import se.sics.kompics.timer.Timer;

import scribe.simulator.snapshot.Snapshot;
import tman.system.peer.tman.TManPartnersPort;
import tman.system.peer.tman.TManPartnersRequest;
import tman.system.peer.tman.TManPartnersResponse;

public final class Scribe extends ComponentDefinition {
	Negative<ScribePort> scribePort = negative(ScribePort.class);
	Positive<TManPartnersPort> tmanPartnersPort = positive(TManPartnersPort.class);
	Positive<Network> networkPort = positive(Network.class);
	Positive<Timer> timerPort = positive(Timer.class);

	private long period;
	private PeerAddress self;
	private ArrayList<BigInteger> topics;
	private ArrayList<PeerAddress> tmanPartners;
	
	private Map<BigInteger, Set<PeerAddress>> forwardedTopics = new HashMap<BigInteger, Set<PeerAddress>>();
	private Set<BigInteger> rootedTopics = new HashSet<BigInteger>(); 
//-------------------------------------------------------------------	
	public Scribe() {
		topics = new ArrayList<BigInteger>();
		tmanPartners = new ArrayList<PeerAddress>();
		
		subscribe(handleInit, control);
		subscribe(handleRequestTManPartners, timerPort);
		subscribe(handleRecvTopicEvent, networkPort);
		subscribe(handleRecvTManPartners, tmanPartnersPort);
		subscribe(handlePublish, scribePort);
		subscribe(handleScribeJoinRequest, networkPort);
	}

//-------------------------------------------------------------------	
	Handler<ScribeInit> handleInit = new Handler<ScribeInit>() {
		public void handle(ScribeInit init) {
			self = init.getSelf();
			topics = init.getTopics();
			period = init.getPeriod();
			
			SchedulePeriodicTimeout rst = new SchedulePeriodicTimeout(period, period);
			rst.setTimeoutEvent(new ScribeSchedule(rst));
			trigger(rst, timerPort);
		}
	};

//-------------------------------------------------------------------	
	Handler<PublishEvent> handlePublish = new Handler<PublishEvent>() {
		public void handle(PublishEvent event) {
			BigInteger topicId = event.getTopicId();
			Snapshot.publishTopicEvent(topicId, self);
			System.out.println("Children of root "+self+" for topic "+event.getTopicId());
			if(forwardedTopics.get(event.getTopicId()) != null){
				for(PeerAddress child : forwardedTopics.get(event.getTopicId())){
					System.out.print(child+" ");
				}
				System.out.println();
			}
			
			if(forwardedTopics.containsKey(topicId)){
				forwardToChildren(topicId);
			}
		}
	};

//-------------------------------------------------------------------	
	Handler<TopicEvent> handleRecvTopicEvent = new Handler<TopicEvent>() {
		public void handle(TopicEvent event) {
			BigInteger topicId = event.getTopicId();
			Snapshot.updateRecvTopicEvent(self, topicId);
			if(forwardedTopics.containsKey(topicId)){
				forwardToChildren(topicId);
			}
		}
	};

//-------------------------------------------------------------------	
	Handler<ScribeSchedule> handleRequestTManPartners = new Handler<ScribeSchedule>() {
		public void handle(ScribeSchedule event) {
			TManPartnersRequest request = new TManPartnersRequest();
			trigger(request, tmanPartnersPort);
		}
	};

//-------------------------------------------------------------------	
	Handler<TManPartnersResponse> handleRecvTManPartners = new Handler<TManPartnersResponse>() {
		boolean subscribed = false;
		int tmanRounds = 0;

		public void handle(TManPartnersResponse event) {
			tmanPartners = event.getPartners();
			Snapshot.updateTManPartners(self, tmanPartners);
			if(tmanRounds > 20){
				if(! subscribed){
					subscribe(topics);
					subscribed = true;
				}
			} else tmanRounds++;
		}
	};

	private void subscribe(ArrayList<BigInteger> topics) {
		for(BigInteger topic : topics){
			subscribe(topic);
		}
		
	}
	
	private void forwardToChildren(BigInteger topicId) {
		for(PeerAddress child : forwardedTopics.get(topicId)){
			TopicEvent event = new TopicEvent(topicId, self, child);
			trigger(event, networkPort);
		}
	}

	private void subscribe(BigInteger topic) {
		ArrayList<PeerAddress> tmp = new ArrayList<PeerAddress>(tmanPartners);
		tmp.add(self);
		PeerAddress parent = PeerAddress.closestOnARing(tmp, topic);
		if(parent.equals(self)){
			// We are the closest node to the topic, we are the RDV node
			rootedTopics.add(topic);
			Snapshot.rootedTopic(topic, self);
		} else {
			// We are not the closest node to the topic, we must contact the closest node we know of
			ScribeJoinRequest request = new ScribeJoinRequest(self, topic, parent);
			trigger(request, networkPort);
		}
		
	}

	Handler<ScribeJoinRequest> handleScribeJoinRequest = new Handler<ScribeJoinRequest>() {

		@Override
		public void handle(ScribeJoinRequest arg0) {
			
			// If we are already a forwarder for the topic
			if(forwardedTopics.keySet().contains(arg0.getTopic())){
				forwardedTopics.get(arg0.getTopic()).add(arg0.getPeerSource());
			}
			// If we don't know this topic already
			else {
				// Adding the entry in the forwarded topics
				HashSet<PeerAddress> tmp = new HashSet<PeerAddress>();
				tmp.add(arg0.getPeerSource());
				forwardedTopics.put(arg0.getTopic(), tmp);
				
				// Subscribing to the topic;
				subscribe(arg0.getTopic());
			}
		}
	
	
	};
}
